package com.actor;

import java.util.concurrent.TimeUnit;

import com.message.TransformationMessages;

import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.cluster.Cluster;
import akka.cluster.ClusterEvent;
import akka.cluster.ClusterEvent.CurrentClusterState;
import akka.cluster.Member;
import akka.cluster.MemberStatus;
import akka.contrib.pattern.ClusterSingletonProxy;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.pattern.Patterns;
import akka.util.Timeout;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;

public class ClusterServerMasterActor extends UntypedActor {

	LoggingAdapter logger = Logging.getLogger(getContext().system(), this);

	Cluster cluster = Cluster.get(getContext().system());
	
	private static ClusterServerMasterActor INSTANCE;
	
	public ClusterServerMasterActor() {
		INSTANCE = this;
	}
	
	public static ClusterServerMasterActor getInstance(){
		return INSTANCE;
	}
	
	public void sendToGameServer(TransformationMessages.GameServerPacket resPacket){
		CurrentClusterState state = cluster.state();
		for(Member member :state.getMembers()){
			if(member.hasRole(String.valueOf(resPacket.getGameServerId()))){
				resPacket.setSameServer(false);
				ActorSelection  actorSelection  = getContext().actorSelection(member.address() + "/user/myAkkaClusterClient");
				actorSelection.tell(resPacket, getSelf());
				break;
			}
		}
	}
	
	public Future<Object> askToGameServer(TransformationMessages.GameServerPacket resPacket){
		CurrentClusterState state = cluster.state();
		for(Member member :state.getMembers()){
			if(member.hasRole(String.valueOf(resPacket.getGameServerId()))){
				ActorSelection  actorSelection  = getContext().actorSelection(member.address() + "/user/myAkkaClusterClient");
				Future<Object> future = Patterns.ask(actorSelection, resPacket, new Timeout(FiniteDuration.create(5, TimeUnit.SECONDS)));
				return future;
			}
		}
		return null;
	}

	// subscribe to cluster changes
	@Override
	public void preStart() {
		// #subscribe
		cluster.subscribe(getSelf(), ClusterEvent.MemberUp.class);
		// #subscribe
	}

	// re-subscribe when restart
	@Override
	public void postStop() {
		cluster.unsubscribe(getSelf());
	}

	@Override
	public void onReceive(Object message) {
		if (message instanceof TransformationMessages.WorldPacket) {
			TransformationMessages.WorldPacket packet = (TransformationMessages.WorldPacket)message;
			ActorRef ref = getContext().actorOf(Props.create(ClusterServerWorkerActor.class));
			//ActorRef ref = this.getRemoteActor(String.valueOf(packet.getCmd())); // getContext().actorOf(Props.create(ClusterServerWorkerActor.class));
			ref.tell(message, getSender());
		}else if (message instanceof ClusterEvent.CurrentClusterState) {
			/**
			 * 当前节点在刚刚加入集群时，会收到CurrentClusterState消息，从中可以解析出集群中的所有前端节点（即roles为frontend的），并向其发送BACKEND_REGISTRATION消息，用于注册自己
			 */
			ClusterEvent.CurrentClusterState state = (ClusterEvent.CurrentClusterState) message;
			for (Member member : state.getMembers()) {
				if (member.status().equals(MemberStatus.up())) {
					register(member);
				}
			}
		} else if (message instanceof ClusterEvent.MemberUp) {
			/**
			 * 有新的节点加入
			 */
			ClusterEvent.MemberUp mUp = (ClusterEvent.MemberUp) message;
			register(mUp.member());

		}else {
			unhandled(message);
		}

	}

	/**
	 * 如果是客户端角色，则像客户端注册自己的信息。客户端收到消息以后会讲这个服务端存到本机服务列表中
	 * 
	 * @param member
	 */
	void register(Member member) {
		if (member.hasRole("client")) {
			ActorSelection  actorSelection  = getContext().actorSelection(member.address() + "/user/myAkkaClusterClient");
			actorSelection.tell(TransformationMessages.BACKEND_REGISTRATION, getSelf());
		}
	}
	
	private ActorRef getRemoteActor(String id) {
		// int totalInstances = 100;
		// Iterable<String> routeesPaths = Arrays.asList("/user/" + id);
		// boolean allowLocalRoutees = true;
		// ClusterRouterGroup clusterRouterGroup = new ClusterRouterGroup(
		// new AdaptiveLoadBalancingGroup(
		// HeapMetricsSelector.getInstance(),
		// Collections.<String> emptyList()),
		// new ClusterRouterGroupSettings(totalInstances, routeesPaths,
		// allowLocalRoutees, "oschat"));
		// ActorRef remoteActor = getContext().system().actorOf(
		// clusterRouterGroup.props());
		//
		ActorRef remoteActor = getContext().actorOf(
				ClusterSingletonProxy.defaultProps("user/" + id + "/" + id,
						"myAkkaClusterClient"));
		return remoteActor;
	}

}
